use crate::error::{err, Result, StatusCode};
use crate::key_types::InternalKey;
use crate::types::{FileMetaData, FileNum, SequenceNumber};

use integer_encoding::{VarIntReader, VarIntWriter};

use bytes::Bytes;
use std::collections::HashSet;
use std::io::{Read, Write};

#[derive(PartialEq, Debug, Clone)]
pub struct CompactionPointer {
    pub level: usize,
    // This key is in InternalKey format.
    pub key: Bytes,
}

enum EditTag {
    Comparator = 1,
    LogNumber = 2,
    NextFileNumber = 3,
    LastSequence = 4,
    CompactPointer = 5,
    DeletedFile = 6,
    NewFile = 7,
    PrevLogNumber = 9, // sic!
}

fn tag_to_enum(t: u32) -> Option<EditTag> {
    match t {
        1 => Some(EditTag::Comparator),
        2 => Some(EditTag::LogNumber),
        3 => Some(EditTag::NextFileNumber),
        4 => Some(EditTag::LastSequence),
        5 => Some(EditTag::CompactPointer),
        6 => Some(EditTag::DeletedFile),
        7 => Some(EditTag::NewFile),
        9 => Some(EditTag::PrevLogNumber),
        _ => None,
    }
}

fn read_length_prefixed<R: Read>(reader: &mut R) -> Result<Vec<u8>> {
    if let Ok(klen) = reader.read_varint() {
        let mut keybuf = vec![0; klen];

        if let Ok(l) = reader.read(&mut keybuf) {
            if l != klen {
                return err(StatusCode::IOError, "Couldn't read full key");
            }
            Ok(keybuf)
        } else {
            err(StatusCode::IOError, "Couldn't read key")
        }
    } else {
        err(StatusCode::IOError, "Couldn't read key length")
    }
}

/// Manages changes to the set of managed SSTables and logfiles.
pub struct VersionEdit {
    /// 比较器
    comparator: Option<String>,
    /// 日志文件编号
    pub log_number: Option<FileNum>,
    /// 前一个日志文件编号
    pub prev_log_number: Option<FileNum>,
    /// 后一个日志文件编号
    pub next_file_number: Option<FileNum>,
    /// 最后一个数据编号
    pub last_seq: Option<SequenceNumber>,
    /// 存储了各层级（level）需要更新的压缩指针（key）
    pub compaction_ptrs: Vec<CompactionPointer>,
    /// 要删除的文件编号集合
    pub deleted: HashSet<(usize, FileNum)>,
    /// 新开一个文件
    pub new_files: Vec<(usize, FileMetaData)>,
}

impl VersionEdit {
    pub fn new() -> VersionEdit {
        VersionEdit {
            comparator: None,
            log_number: None,
            prev_log_number: None,
            next_file_number: None,
            last_seq: None,
            compaction_ptrs: Vec::with_capacity(8),
            deleted: HashSet::with_capacity(8),
            new_files: Vec::with_capacity(8),
        }
    }

    pub fn clear(&mut self) {
        *self = VersionEdit::new();
    }

    /// 这段代码定义了 add_file 方法，用于在版本编辑（VersionEdit）中记录需要添加到指定层级的新 SSTable 文件元数据，是 LSM 树存储引擎中版本更新的重要操作。
    ///
    /// 具体功能分析：
    ///
    /// 方法接收两个参数：level（新文件要添加到的目标层级）和 file（新 SSTable 的元数据 FileMetaData，包含文件编号、键范围、大小等信息）。
    ///
    /// 它将这两个参数组成的元组 (level, file) 添加到 self.new_files 列表中（new_files 通常是一个存储新增文件信息的向量）。
    ///
    /// 在 LSM 树中，这个方法的典型使用场景是：
    ///
    /// 压缩（compaction）操作生成新的 SSTable 后，通过 add_file 将这些新文件记录到版本编辑中，准备添加到目标层级。
    ///
    /// 内存表（MemTable）刷写到磁盘形成新的 SSTable 后，将其添加到 L0 层时。
    ///
    /// 这些新增文件的记录会在版本更新时（将 VersionEdit 应用到当前版本）生效，与 delete_file 方法配合，共同完成旧文件移除和新文件添加的版本迭代，确保数据的一致性和正确性。
    ///
    /// 简单来说，它是一个 “文件添加记录器”，用于记录哪些新的 SSTable 文件需要添加到哪个层级，是存储引擎中数据持久化和版本演进的关键步骤。
    pub fn add_file(&mut self, level: usize, file: FileMetaData) {
        self.new_files.push((level, file))
    }

    /// 这段代码定义了 delete_file 方法，用于在版本编辑（VersionEdit）中记录需要从指定层级删除的 SSTable 文件，是 LSM 树存储引擎中版本管理的重要操作。
    ///
    /// 具体功能分析：
    ///
    /// 方法接收两个参数：level（要删除文件所在的层级）和 file_num（SSTable 的文件编号，唯一标识一个 SSTable）。
    ///
    /// 它将这两个参数组成的元组 (level, file_num) 插入到 self.deleted 集合中（deleted 通常是一个存储待删除文件信息的集合，如 HashSet）。
    ///
    /// 在 LSM 树中，这个方法的作用是标记需要从某个层级移除的 SSTable 文件，通常在以下场景使用：
    ///
    /// 压缩（compaction）操作完成后，参与压缩的旧 SSTable 不再需要，通过 delete_file 标记为待删除。
    ///
    /// 版本迭代过程中，需要移除某些过时或无效的 SSTable 时。
    ///
    /// 标记后的删除操作不会立即执行，而是在版本变更（VersionEdit 应用到当前版本）时统一处理，确保数据一致性和操作原子性。
    ///
    /// 简单来说，它是一个 “文件删除标记器”，用于记录哪些 SSTable 文件需要从哪个层级中删除，为后续的版本更新提供依据。
    pub fn delete_file(&mut self, level: usize, file_num: FileNum) {
        self.deleted.insert((level, file_num));
    }

    pub fn set_comparator_name(&mut self, name: &str) {
        self.comparator = Some(name.to_string())
    }

    pub fn set_log_num(&mut self, num: u64) {
        self.log_number = Some(num)
    }

    pub fn set_prev_log_num(&mut self, num: u64) {
        self.prev_log_number = Some(num);
    }

    pub fn set_last_seq(&mut self, num: u64) {
        self.last_seq = Some(num)
    }

    pub fn set_next_file(&mut self, num: FileNum) {
        self.next_file_number = Some(num)
    }

    /// 这个 set_compact_pointer 方法用于记录 LSM 树中指定层级（level）的压缩进度指针。
    /// 压缩指针（Compaction Pointer）是 LSM 树压缩策略中的关键元数据，用于标记该层级已完成压缩的最大键位置，避免后续压缩操作重复处理已合并的数据，提升压缩效率。
    pub fn set_compact_pointer(&mut self, level: usize, key: InternalKey) {
        // CompactionPointer 结构体：包含两个字段，level 标识所属层级，key 存储该层级已完成压缩的最大内部键（InternalKey）；
        // 数据持久化：key.to_vec().into() 将输入的内部键转换为自有数据（通常是 Vec<u8>），确保指针信息独立存储，不依赖外部数据的生命周期；
        // 存储位置：self.compaction_ptrs 是存储所有层级压缩指针的集合（通常是一个向量），便于后续查询和持久化。
        self.compaction_ptrs.push(CompactionPointer {
            level,  // 记录指针所属的层级
            key: key.to_vec().into(),  // 记录该层级已压缩到的最大内部键（转为自有数据）
        })
    }

    /// Encode this VersionEdit into a buffer.
    pub fn encode(&self) -> Vec<u8> {
        let mut buf = Vec::with_capacity(256);

        if let Some(ref cmp) = self.comparator {
            // swallow errors, because it's a pure in-memory write
            buf.write_varint(EditTag::Comparator as u32).unwrap();
            // data is prefixed by a varint32 describing the length of the following chunk
            buf.write_varint(cmp.len()).unwrap();
            buf.write_all(cmp.as_bytes()).unwrap();
        }

        if let Some(lognum) = self.log_number {
            buf.write_varint(EditTag::LogNumber as u32).unwrap();
            buf.write_varint(lognum).unwrap();
        }

        if let Some(prevlognum) = self.prev_log_number {
            buf.write_varint(EditTag::PrevLogNumber as u32).unwrap();
            buf.write_varint(prevlognum).unwrap();
        }

        if let Some(nfn) = self.next_file_number {
            buf.write_varint(EditTag::NextFileNumber as u32).unwrap();
            buf.write_varint(nfn).unwrap();
        }

        if let Some(ls) = self.last_seq {
            buf.write_varint(EditTag::LastSequence as u32).unwrap();
            buf.write_varint(ls).unwrap();
        }

        for cptr in self.compaction_ptrs.iter() {
            buf.write_varint(EditTag::CompactPointer as u32).unwrap();
            buf.write_varint(cptr.level).unwrap();
            buf.write_varint(cptr.key.len()).unwrap();
            buf.write_all(cptr.key.as_ref()).unwrap();
        }

        for df in self.deleted.iter() {
            buf.write_varint(EditTag::DeletedFile as u32).unwrap();
            buf.write_varint(df.0).unwrap();
            buf.write_varint(df.1).unwrap();
        }

        for nf in self.new_files.iter() {
            buf.write_varint(EditTag::NewFile as u32).unwrap();
            buf.write_varint(nf.0).unwrap();
            buf.write_varint(nf.1.num).unwrap();
            buf.write_varint(nf.1.size).unwrap();

            buf.write_varint(nf.1.smallest.len()).unwrap();
            buf.write_all(nf.1.smallest.as_ref()).unwrap();
            buf.write_varint(nf.1.largest.len()).unwrap();
            buf.write_all(nf.1.largest.as_ref()).unwrap();
        }

        buf
    }

    pub fn decode_from(src: &[u8]) -> Result<VersionEdit> {
        let mut reader = src;
        let mut ve = VersionEdit::new();

        while let Ok(tag) = reader.read_varint::<u32>() {
            if let Some(tag) = tag_to_enum(tag) {
                match tag {
                    EditTag::Comparator => {
                        let buf = read_length_prefixed(&mut reader)?;
                        if let Ok(c) = String::from_utf8(buf) {
                            ve.comparator = Some(c);
                        } else {
                            return err(StatusCode::Corruption, "Bad comparator encoding");
                        }
                    }

                    EditTag::LogNumber => {
                        if let Ok(ln) = reader.read_varint() {
                            ve.log_number = Some(ln);
                        } else {
                            return err(StatusCode::IOError, "Couldn't read lognumber");
                        }
                    }

                    EditTag::PrevLogNumber => {
                        if let Ok(ln) = reader.read_varint() {
                            ve.prev_log_number = Some(ln);
                        } else {
                            return err(StatusCode::IOError, "Couldn't read prevlognumber");
                        }
                    }

                    EditTag::NextFileNumber => {
                        if let Ok(nfn) = reader.read_varint() {
                            ve.next_file_number = Some(nfn);
                        } else {
                            return err(StatusCode::IOError, "Couldn't read next_file_number");
                        }
                    }

                    EditTag::LastSequence => {
                        if let Ok(ls) = reader.read_varint() {
                            ve.last_seq = Some(ls);
                        } else {
                            return err(StatusCode::IOError, "Couldn't read last_sequence");
                        }
                    }

                    EditTag::CompactPointer => {
                        // Monads by indentation...
                        if let Ok(lvl) = reader.read_varint() {
                            let key = read_length_prefixed(&mut reader)?.into();

                            ve.compaction_ptrs
                                .push(CompactionPointer { level: lvl, key });
                        } else {
                            return err(StatusCode::IOError, "Couldn't read level");
                        }
                    }

                    EditTag::DeletedFile => {
                        if let Ok(lvl) = reader.read_varint() {
                            if let Ok(num) = reader.read_varint() {
                                ve.deleted.insert((lvl, num));
                            } else {
                                return err(StatusCode::IOError, "Couldn't read file num");
                            }
                        } else {
                            return err(StatusCode::IOError, "Couldn't read level");
                        }
                    }

                    EditTag::NewFile => {
                        if let Ok(lvl) = reader.read_varint() {
                            if let Ok(num) = reader.read_varint() {
                                if let Ok(size) = reader.read_varint() {
                                    let smallest = read_length_prefixed(&mut reader)?.into();
                                    let largest = read_length_prefixed(&mut reader)?.into();
                                    ve.new_files.push((
                                        lvl,
                                        FileMetaData {
                                            num,
                                            size,
                                            smallest,
                                            largest,
                                            allowed_seeks: 0,
                                        },
                                    ))
                                } else {
                                    return err(StatusCode::IOError, "Couldn't read file size");
                                }
                            } else {
                                return err(StatusCode::IOError, "Couldn't read file num");
                            }
                        } else {
                            return err(StatusCode::IOError, "Couldn't read file level");
                        }
                    }
                }
            } else {
                return err(
                    StatusCode::Corruption,
                    &format!("Invalid tag number {}", tag),
                );
            }
        }

        Ok(ve)
    }
}

impl Default for VersionEdit {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::CompactionPointer;
    use super::VersionEdit;

    use crate::cmp::{Cmp, DefaultCmp};
    use crate::types::FileMetaData;

    #[test]
    fn test_version_edit_encode_decode() {
        let mut ve = VersionEdit::new();

        ve.set_comparator_name(DefaultCmp.id());
        ve.set_log_num(123);
        ve.set_next_file(456);
        ve.set_compact_pointer(0, &[0, 1, 2]);
        ve.set_compact_pointer(1, &[3, 4, 5]);
        ve.set_compact_pointer(2, &[6, 7, 8]);
        ve.add_file(
            0,
            FileMetaData {
                allowed_seeks: 12345,
                num: 901,
                size: 234,
                smallest: vec![5, 6, 7].into(),
                largest: vec![8, 9, 0].into(),
            },
        );
        ve.delete_file(1, 132);

        let encoded = ve.encode();

        let decoded = VersionEdit::decode_from(encoded.as_ref()).unwrap();

        assert_eq!(decoded.comparator, Some(DefaultCmp.id().to_string()));
        assert_eq!(decoded.log_number, Some(123));
        assert_eq!(decoded.next_file_number, Some(456));
        assert_eq!(decoded.compaction_ptrs.len(), 3);
        assert_eq!(
            decoded.compaction_ptrs[0],
            CompactionPointer {
                level: 0,
                key: vec![0, 1, 2].into(),
            }
        );
        assert_eq!(
            decoded.compaction_ptrs[1],
            CompactionPointer {
                level: 1,
                key: vec![3, 4, 5].into(),
            }
        );
        assert_eq!(
            decoded.compaction_ptrs[2],
            CompactionPointer {
                level: 2,
                key: vec![6, 7, 8].into(),
            }
        );
        assert_eq!(decoded.new_files.len(), 1);
        assert_eq!(
            decoded.new_files[0],
            (
                0,
                FileMetaData {
                    allowed_seeks: 0,
                    num: 901,
                    size: 234,
                    smallest: vec![5, 6, 7].into(),
                    largest: vec![8, 9, 0].into(),
                }
            )
        );
        assert_eq!(decoded.deleted.len(), 1);
        assert!(decoded.deleted.contains(&(1, 132)));
    }
}
